package com.dhz.common.queue.guava;

import com.google.common.eventbus.AsyncEventBus;

import javax.annotation.PreDestroy;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

public abstract class AbstractGuavaQueue<E> {

    private AsyncEventBus asyncEventBus;
    private String threadId = "default";
    private int threadNum = 3;
    private boolean initialized = false;
    private E event;

    public void setThreadId(String threadId) {
        this.threadId = threadId;
    }

    public void setThreadNum(int threadNum) {
        this.threadNum = threadNum;
    }

    public abstract void init() throws Exception;

    @PreDestroy
    public void unRegister() {
        if (asyncEventBus != null && event != null) {
            asyncEventBus.unregister(event);
        }
        GuavaQueueManager.shutdown(threadId);
    }

    public void post(Object message) {
        if (!initialized) {
            try {
                register();
            } catch (Exception e) {
                throw new GuavaQueueException(e);
            }
        }
        asyncEventBus.post(message);
    }

    @SuppressWarnings("unchecked")
    public void register() throws Exception {

        Type type = ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        event = ((Class <E>)type).newInstance();

        if (event == null) {
            throw new GuavaQueueException("Queue Message Sub object is must.");
        }

        if (threadNum < 1 || threadNum > 20) {
            throw new GuavaQueueException("thread num must between 1 and 20.");
        }
        asyncEventBus = GuavaQueueManager.createAsyncEventBus(threadId, threadNum);
        asyncEventBus.register(event);
        this.initialized = true;

    }
}
